package com.lvhx.springboot.partitioner;

import com.thoughtworks.xstream.XStream;
import com.thoughtworks.xstream.io.json.JettisonMappedXmlDriver;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component("dataCenter")
public class SimpleDataCenter implements DataCenter<Object> {
	private static final String SAVE_DATA = "INSERT into BATCH_DATA_CENTER(DATA_KEY, SERIALIZED_DATA, DATA_COUNT, SERIAL_DURATIME, CREATE_DATETIME)"
			+ "values(?, ? ,? ,? ,?)";
	private static final String GET_DATA = "SELECT DATA_KEY, SERIALIZED_DATA, DATA_COUNT, SERIAL_DURATIME,PERSIST_DURATIME from BATCH_DATA_CENTER where DATA_KEY = ?";

	private static final String UPDATE_PERSIST_DURATIME = "UPDATE BATCH_DATA_CENTER SET PERSIST_DURATIME = ? WHERE DATA_KEY = ?";

	private Map<String, List<Object>> dataMap = new ConcurrentHashMap<String, List<Object>>();

	private JdbcTemplate JdbcTemplate;

	private XStream xstream = new XStream(new JettisonMappedXmlDriver());

	@Autowired
	public SimpleDataCenter(DataSource batchDataSource) {
		this.JdbcTemplate = new JdbcTemplate(batchDataSource);
	}

	@Override
	public List<Object> getData(String key) {
		if (dataMap.containsKey(key)) {
			return dataMap.get(key);
		} else {
			return retrive(key);
		}
	}

	@Override
	public void saveInMemory(String key, List<Object> data) {
		dataMap.put(key, data);
	}

	@Override
	public void saveInDataStore(String key, List<Object> data) {
		persist(key, data);
	}

	@Override
	public void saveInMemoryAndDataStore(String key, List<Object> data) {
		dataMap.put(key, data);
		persist(key, data);
	}

	@Override
	public void removeIfDataInMemory(String key) {
		if (dataMap.containsKey(key)) {
			dataMap.remove(key);
		}
	}

	private void persist(String key, List<Object> data) {
		long startTime = 0;
		long duraTime = 0;

		// 序列化
		startTime = System.currentTimeMillis();
		String serializedData = xstream.toXML(data);
		duraTime = System.currentTimeMillis() - startTime;

		// 持久化
		startTime = System.currentTimeMillis();
		Object[] parameters = new Object[] { key, serializedData, data.size(),
				duraTime, new Date() };
		JdbcTemplate.update(SAVE_DATA, parameters, new int[] { Types.VARCHAR,
				Types.CLOB, Types.NUMERIC, Types.NUMERIC, Types.TIMESTAMP });
		duraTime = System.currentTimeMillis() - startTime;

		// 更新持久化所用时间
		JdbcTemplate.update(UPDATE_PERSIST_DURATIME, new Object[] { duraTime,
				key }, new int[] { Types.NUMERIC, Types.VARCHAR });

	}

	private List<Object> retrive(String key) {
		DataItem item = (DataItem) JdbcTemplate.queryForObject(GET_DATA,
				new Object[] { key }, new DataItemMapper());
		if (item != null) {
			return item.getData();
		} else {
			throw new RuntimeException("there is no data with partkey " + key);
		}
	}

	private class DataItemMapper implements RowMapper<DataItem> {
		@SuppressWarnings("unchecked")
		public DataItem mapRow(ResultSet rs, int i) throws SQLException {
			DataItem item = new DataItem();
			item.setDataKey(rs.getString("DATA_KEY"));
			item.setDataCount(rs.getLong("DATA_COUNT"));
			item.setSerialDuraTime(rs.getLong("SERIAL_DURATIME"));
			item.setPersistDuraTime(rs.getLong("PERSIST_DURATIME"));

			List<Object> data = (List<Object>) xstream.fromXML(rs
					.getString("SERIALIZED_DATA"));
			item.setData(data);

			return item;
		}
	}

	private class DataItem {
		private String dataKey;
		private List<Object> data;
		private long dataCount;
		private long serialDuraTime;
		private long persistDuraTime;

		public String getDataKey() {
			return dataKey;
		}

		public void setDataKey(String dataKey) {
			this.dataKey = dataKey;
		}

		public List<Object> getData() {
			return data;
		}

		public void setData(List<Object> data) {
			this.data = data;
		}

		public long getDataCount() {
			return dataCount;
		}

		public void setDataCount(long dataCount) {
			this.dataCount = dataCount;
		}

		public long getSerialDuraTime() {
			return serialDuraTime;
		}

		public void setSerialDuraTime(long serialDuraTime) {
			this.serialDuraTime = serialDuraTime;
		}

		public long getPersistDuraTime() {
			return persistDuraTime;
		}

		public void setPersistDuraTime(long persistDuraTime) {
			this.persistDuraTime = persistDuraTime;
		}
	}
}
